package com.heima.kafkastream;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.apache.kafka.streams.state.KeyValueStore;

import java.util.Arrays;
import java.util.Properties;

public class WordCountApplication {

   public static void main(final String[] args) throws Exception {
       //1.构建属性对象
       Properties props = new Properties();
       //1.1 设置应用ID
       props.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application");
       //1.2 设置broker服务器地址
       props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.137.136:9092");
       //1.3 设置Key的序列化器     从某一个原始话题中获取数据，取到数据的 key value 则要指定序列化器
       props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
       //1.4 设置val的序列化器
       props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());


       //2. 创建流构建对象
       StreamsBuilder builder = new StreamsBuilder();
       //2.1 设置源头话题名称-----源处理器   --TODO 需要手动先创建该原始话题  需要kafkafrop界面即192.168.137.136:9001手动创建 点new  topic name写此TextLinesTopic但此时话题是空的 还需要有家人发消息so点从外部模块到点pom就导入了让kafka-demo的ProducerTest修改topic成此话题  message是多个评价就按空格分割 向原始话题发数据     再启动此代码否则报错
       KStream<String, String> textLines = builder.stream("TextLinesTopic");//数据源源不断来到原始话题


       //2.2 流式计算逻辑 ---  流拓扑结构中-------流处理器
       KTable<String, Long> wordCounts = textLines
               //2.2.1 处理数据Val中的值  -将英文转为小写 采用空格分隔
           .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
               //2.2.2 对处理后数据进行分组
           .groupBy((key, word) -> word)
               //2.2.3 对处理后数据进行统计个数
           .count(Materialized.<String, Long, KeyValueStore<Bytes, byte[]>>as("counts-store"));
       //2.3 设置目标话题-----下沉处理器        kafka内部对数据进行逻辑运算，计算后的结果被实时发到了此目标话题
       wordCounts.toStream().to("WordsWithCountsTopic", Produced.with(Serdes.String(), Serdes.Long()));
//                                   kafkadrop显示的就是此目标话题字母
       //3.构建流式应用           普通消费者进行消费
       KafkaStreams streams = new KafkaStreams(builder.build(), props);

       //4.启动流式应用开始计算             将结果展示到直播间
       streams.start();
   }

}
